import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.io.BufferedReader;
import java.io.FileNotFoundException;
import java.io.FileReader;
import java.util.Properties;

public class Demo02KafkaCarProducer {
    public static void main(String[] args) throws Exception {
        // 读取文件数据，将每一条数据逐条写入Kafka中
        BufferedReader bs = new BufferedReader(new FileReader("flink/data/cars_sample.txt"));

        String line = bs.readLine();

        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "master:9092,node2:9092,node2:9092");

        properties.setProperty("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.setProperty("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");

        KafkaProducer<String, String> producer = new KafkaProducer<>(properties);

        int cnt = 0;
        while (line != null) {
            producer.send(new ProducerRecord<>("cars_kafka", line));
            producer.flush();
            // 每条数据暂停一会儿
//            Thread.sleep(100);
            cnt++;
            System.out.println("写入了" + cnt + "条数据");
            line = bs.readLine();
        }

        bs.close();
        producer.close();


    }
}
